// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jing

#include "coin2/exchange/quoinex/feed_v2/processor.h"

#include <string>
#include <utility>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/quoinex/feed_v2/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::quoinex::feed_v2::impl {

static const std::string kBook = "price_ladders_cash";
static const std::string kTrade = "executions_cash";

struct QuoinexChannel {
  std::string type;
  std::string symbol;

  static QuoinexChannel Parse(const std::string& ch) {
    /*
     * Book channel:  price_ladders_cash_SYMBOL
     * Trade channel: executions_cash_SYMBOL
     */
    std::vector<std::string> splitted;
    splitted = Split(ch, "_");
    if (splitted.size() == 4) {
      return QuoinexChannel{kBook, splitted[3]};
    } else if (splitted.size() == 3) {
      return QuoinexChannel{kTrade, splitted[2]};
    } else {
      THROW() << "[Quoinex Feed Processor] Unknown channel: " << ch;
    }
    return QuoinexChannel();
  }
};

void QuoinexFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: " << GetParseError_En(document.GetParseError());
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (!document.HasMember("event")) {
    LOG(INFO) << "[Quoinex Feed Processor] unknown msg for " << packet.topic_name << ": "
              << packet.payload;
    return;
  }

  const std::string& event = document["event"].GetString();
  if (event == "created" || event == "updated") {
    auto channel = QuoinexChannel::Parse(document["channel"].GetString());

    if (!CheckChannelFilter(channel.type)) {
      return;
    }

    if (!document.HasMember("data")) {
      LOG(INFO) << "[Quoinex Feed Processor] unknown msg for " << packet.topic_name << ": "
                << packet.payload;
      return;
    }
    const std::string& data_str = document["data"].GetString();
    rapidjson::Document data;
    data.Parse<rapidjson::kParseNumbersAsStringsFlag>(data_str.data());
    if (data.HasParseError()) {
      THROW() << "JSON parse error: " << GetParseError_En(data.GetParseError());
    }

    const std::string& native_symbol = channel.symbol;
    if (channel.type == kTrade) {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }
      Trade* trade = upd_mgr_.GetTradeMutable();
      QuoinexFeedUpdateParser::ParseTrade(data, trade, packet.timestamp);
      upd_mgr_.PublishTrade(native_symbol, false);
    } else if (channel.type == kBook) {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        QuoinexFeedUpdateParser::ParseBook(book, data);
        upd_mgr_.PublishBook();
      }
    } else {
      LOG(INFO) << "[Quoinex Feed Processor] unknown msg for " << packet.topic_name << ": "
                << packet.payload;
    }
  } else if (event != "pusher:pong") {
    // meta msg like subscription response...
    LOG(INFO) << "[Quoinex Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload;
  }

  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::quoinex::feed_v2::impl
